#set( $symbol_pound = '#' )
#set( $symbol_dollar = '$' )
#set( $symbol_escape = '\' )
package ${package}.config.mqtt;

import com.fasterxml.jackson.databind.ObjectMapper;
import ${package}.config.mqtt.property.MqttProperty;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@ConditionalOnBean(MqttConfig.class)
public class MqttConsumerConfig {
    private final ObjectMapper objectMapper = new ObjectMapper();
    /**
     * mqtt消费端这种方式不是必须，可以通过注解消费
     *
     * @param mqttClientFactory
     * @return
     */
    @Bean
    public IntegrationFlow mqttInboundFlow(MqttPahoClientFactory mqttClientFactory, MqttMessageConverter messageConverter, MqttProperty mqttProperty, @Qualifier("taskScheduler") TaskScheduler taskScheduler) {
        return IntegrationFlows
                .from(mqttInboundAdapter(mqttClientFactory, mqttProperty, taskScheduler, messageConverter))
//                .transform((GenericTransformer<String, ?>) source -> {
//                    try {
//                        MqttMsgContent<?> mqttMsgContent = objectMapper.readValue(source, MqttMsgContent.class);
//                        return mqttMsgContent.getPayload();
//                    } catch (JsonProcessingException e) {
//                        throw new RuntimeException("Failed to convert JSON to MQTT message", e);
//                    }
//                })
                .get();
    }

    /**
     * mqtt消费端适配器,不使用IntegrationFlow进行消费，可以结合注解消费@ServiceActivator(inputChannel = "mqttInputExecutorChannel")
     *
     * @param mqttClientFactory
     * @return
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInboundAdapter(MqttPahoClientFactory mqttClientFactory, MqttProperty mqttProperty, @Qualifier("taskScheduler") TaskScheduler taskScheduler, MqttMessageConverter messageConverter) {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperty.getClientId()+"-consumer",
                mqttClientFactory,
                "device-mqtt");
        adapter.setTaskScheduler(taskScheduler);
        adapter.setCompletionTimeout(5000);
//        JsonPahoMessageConverter converter = new JsonPahoMessageConverter();
//        adapter.setConverter(converter);
        adapter.setQos(1);
        //手动ack
        adapter.setManualAcks(true);
        adapter.setOutputChannel(mqttInputExecutorChannel());
        //错误处理，重试机制用完还是失败的消息，连接异常
        adapter.setErrorChannel(mqttInputErrorChannel());
        return adapter;
    }

    @Bean
    public MessageChannel mqttInputDirectChannel() {
        return new DirectChannel();
    }

    /**
     * 支持多线程消费
     *
     * @return
     */
    @Bean
    public MessageChannel mqttInputExecutorChannel() {
        return new ExecutorChannel(taskExecutor());
    }

    @Bean
    public MessageChannel mqttInputErrorChannel() {
        return new ExecutorChannel(taskErrorExecutor());
    }

    @Bean
    public ThreadPoolTaskExecutor taskErrorExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("MQTT-Error-Consumer-");
        executor.initialize();
        return executor;
    }

    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("MQTT-Consumer-");
        executor.initialize();
        return executor;
    }

    @Bean
    public RequestHandlerRetryAdvice retryAdvice() {
        RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();

        RetryTemplate retryTemplate = new RetryTemplate();

        // 定义重试策略
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        // 定义回退策略
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(2000L);

        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        advice.setRetryTemplate(retryTemplate);
        return advice;
    }
}
